package com.dove.config.rabbit;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;

/**
 * 为了启用 @RabbitListener 注解，需要在你的某个@Configuration类中添加@EnableRabbit 注解．
 * 
 * @author gaoyibing
 */
@Configuration
@EnableRabbit
public class MsgListenerConfig implements RabbitListenerConfigurer {

	/** 队列消费配置 **/

	@Value("${spring.rabbitmq.host}")
	private String mqRabbitHost;
	@Value("${spring.rabbitmq.port}")
	private int mqRabbitPort;
	@Value("${spring.rabbitmq.username}")
	private String mqRabbitUserName;
	@Value("${spring.rabbitmq.password}")
	private String mqRabbitPassword;
	@Value("${spring.rabbitmq.virtual-host}")
	private String virtualHost;

	@Bean
	public ConnectionFactory connectionFactory() {
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(this.mqRabbitHost, this.mqRabbitPort);
		connectionFactory.setUsername(this.mqRabbitUserName);
		connectionFactory.setPassword(this.mqRabbitPassword);
		connectionFactory.setVirtualHost(this.virtualHost);
		connectionFactory.setChannelCacheSize(100);
		// connectionFactory.setPublisherConfirms(true);
		return connectionFactory;
	}

	@Bean
	public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
		DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
		factory.setMessageConverter(new MappingJackson2MessageConverter());
		return factory;
	}

	// https://docs.spring.io/spring-amqp/docs/1.4.5.RELEASE/reference/html/amqp.html
	@Bean
	public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(@Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);
		// 使用核心轮询大小为3个线程最大10个线程的池大小来调用.
		factory.setConcurrentConsumers(3);
		factory.setMaxConcurrentConsumers(10); // todo thread executor
		// MANUAL 需要手动标记数据是否被成功消费
		factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

		// factory.setPrefetchCount(); 消费者本地的block queue中一次存多少
		return factory;
	}

	@Override
	public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
		registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
	}

}